package com.bungarus.ratelimit;

import com.bungarus.group.cachestore.CacheStore;
import com.bungarus.group.cachestore.CacheStoreCallback;
import com.bungarus.group.cachestore.CacheTransformer;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.TransactionResult;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.sync.RedisCommands;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.time.temporal.TemporalAccessor;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Rate limit judge
 * Created by tang on 2019/10/31.
 */
@Component
public class RateLimitManager {
    private RateLimitCacheTransformer transformer;
    private CacheStore cacheStore;

    @Autowired
    public RateLimitManager(RateLimitCacheTransformer transformer, CacheStore cacheStore) {
        this.transformer = transformer;
        this.cacheStore = cacheStore;
    }

    /**
     * Judge whether client request accept or not. The smallest unit is {@link LimitUnit}.MINUTE.
     * To improve the accurate, we count requests from each sender using multiple fixed time windows 1/60th the size of our rate limit’s time window.
     * @param request client request
     * @return -1 presents ignored, positive presents accepted and the available slot for next request.
     */
    public int pass(final String request, int limit, LimitUnit unit) {
        int cLimit = 0;
        switch (unit) {
            case DAY:
                cLimit = limit / 24 / 60 ;
                break;
            case HOUR:
                cLimit = limit / 60;
                break;
            case MINUTE:
                cLimit = limit;
                break;
            default:
                break;
        }
        cLimit = (0 == cLimit) ? 1 : cLimit;
        Object or = cacheStore.execSyncCommandsWithCallback(CacheTransformer.CACHE_ACTION_HSET, new CacheStoreCallback<RedisCommands<String, String>, List<Integer>>() {
            @Override
            public List<Integer> callHset(RedisCommands<String, String> syncCommand) {
                //write the request
                long timestampSec = System.currentTimeMillis() / 1000L; //transform the milliseconds return to seconds
                long timestampMin = timestampSec / 60; //transform the seconds to minutes
                long timestampHour = timestampMin / 60; //transform the minutes to hours
                long timestampDay = timestampHour / 24; //transform the hours to days

                String key = CacheStore.ofUniqueId(transformer.getCacheTableName(), request);
                String field = String.valueOf(timestampMin);

                syncCommand.multi(); //prepare transaction command
                switch (unit) {
                    case DAY:
                        syncCommand.pexpireat(key, ((timestampDay + 1) * 24 + timestampHour % 24) * 60 * 60 * 1000);   //index 0
                        break;
                    case HOUR:
                        syncCommand.pexpireat(key, ((timestampHour + 1) * 60 + timestampMin % 60)  * 60 * 1000);   //index 0
                        break;
                    case MINUTE:
                        syncCommand.pexpireat(key, ((timestampMin + 1) * 60 + timestampSec % 60)* 1000); //index 0
                        break;
                    default:
                        break;
                }
                syncCommand.hincrby(key, field, 1);   //index 1
                //read the total count, and see if the total request number exceeds the limit or not?
                syncCommand.hgetall(key); //index 2
                TransactionResult transactionResult = syncCommand.exec(); //exec transaction

                Long curFieldTotal = transactionResult.get(1);
                Map<String, String> allRequests = transactionResult.get(2);

                int totalCount = allRequests.values().stream().mapToInt(Integer::parseInt).sum();
                return Arrays.asList(curFieldTotal.intValue(), totalCount);
            }

            @Override
            public List<Integer> callZset(RedisCommands<String, String> syncCommand) {
                throw new UnsupportedOperationException("Method call is not supported.");
            }
        });
        List<Integer> result = (List<Integer>)or;
        if(result.size() == 2) {
            if(result.get(0) > cLimit || result.get(1) > limit) {
                return -1;
            }
            return limit - result.get(1);
        }
        return -1;
    }
}
